package com.study.bigdata.spark.core.rdd.operator.transform

import org.apache.spark.{SparkConf, SparkContext}

object Scala21_RDD_Rep{
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(conf)
    // TODO 统计出每一个省份每个广告被点击数量排行的Top3
    // TODO 1.读取数据文件，获取原始数据
    // 1516609143867 6 7 64 16
    val lines = sc.textFile("data/agent.log")
    // TODO 2.将原始数据进行结构转换
    // line => ((省份，广告),1)
    val wordToOne = lines.map(
      line => {
        val data = line.split(" ")
        ((data(1), data(4)), 1)
      }
    )
    // TODO 3.将转变后的数据进行统计
    // line => ((省份，广告),1) => ((省份，广告),sum)
    val wordToSum = wordToOne.reduceByKey(_+_)
    // TODO 4.将统计结果进行结构转换，将省份独立出来
    // ((省份，广告),sum) => (省份(广告,sum))
    val wordTuple = wordToSum.map {
      case ((prv, adv), sum) => {
        (prv, (adv, sum))
      }
    }
    // TODO 5.按照省份进行分组
    val groupRDD = wordTuple.groupByKey()
    // TODO 6.将分组后的数据进行排行（降序，去前三）
    val top3 = groupRDD.mapValues(
      iter => {
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(3)
      }
    )
    // TODO 7.将结果打印
    top3.collect().foreach(println)
    //(4,List((12,25), (2,22), (16,22)))......

    sc.stop()

  }

}
